package com.intct.func;

import com.alibaba.fastjson.JSONObject;
import com.intct.util.HBaseUtil;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.Map;

/**
 * @author gufg
 * @since 2025-07-03 15:47
 * 自定义HBase Sink
 */
public class HBasekSinkFunc implements SinkFunction<JSONObject> {
    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        // 样例：{"code":110101,"name":"北京市东城区", "op":"r", "sinkTable":"tableName", "sinkFamily:"", "rowkey":""}

        // hbase shell : put 't1', '1001', 'f1:name', 'value'
        // sink表名称
        String sinkTable = jsonObj.getString("sinkTable");
        jsonObj.remove("sinkTable");

        // op操作类型
        String op = jsonObj.getString("op");
        jsonObj.remove("op");

        // rowkey字段名称
        String sinkRowKey = jsonObj.getString("sinkRowKey");
        jsonObj.remove("sinkRowKey");
        String rowkey = jsonObj.getString(sinkRowKey);

        // family
        String sinkFamily = jsonObj.getString("sinkFamily");
        jsonObj.remove("sinkFamily");

        // 最后样例数据：{"code":110101,"name":"北京市东城区"}
        // 是删除 ，在hbase中删除 数据  hbase shell: delete 't1', '1001'
        if ("d".equals(op)) {
            HBaseUtil.delete(sinkTable, rowkey);
            System.out.println("删除数据成功......");
        } else {
            // put实例
            Put put = new Put(Bytes.toBytes(rowkey));

            // 循环将JSON中的Key 和 value取，组合为put所需的列族和列限定符
            for (Map.Entry<String, Object> entry : jsonObj.entrySet()) {
                put.addColumn(Bytes.toBytes(sinkFamily), Bytes.toBytes(entry.getKey()), Bytes.toBytes(String.valueOf(entry.getValue())));
            }
            // 调用插入数据
            HBaseUtil.put(sinkTable, put);
            System.out.println("插入数据成功......");
        }
    }
}
